package com.xzr.demo.mqserver.core;

import com.xzr.demo.common.Consumer;
import com.xzr.demo.common.ConsumerEnv;
import com.xzr.demo.common.MqException;
import com.xzr.demo.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 通过这个类，来实现消费者消息的核心逻辑
 *
 * @Author：徐志荣
 * @Date：2023/8/16 上午 11:24
 */
public class ConsumerManager {
    //持有上层 VirtualHost 对象的引用，用来操作数据
    private VirtualHost parent;
    //指定一个线程池，负责去执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    //存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p) {
        parent = p;
        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    //1. 拿到令牌
                    String queueName = tokenQueue.take();
                    //2. 根据令牌找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后，发现该队列不存在 ， queueName =  " + queueName);
                    }
                    //3. 从这个队列中消费一个消息
                    synchronized (queue) {
                        consumerMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }

            }
        });
        //把线程设为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    /**
     * 这个方法的调用时机 就是 发送消息的时候
     *
     * @param queueName
     * @return
     */
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    /**
     * 添加consumer
     *
     * @param consumerTag
     * @param queueName
     * @param autoAck
     * @param consumer
     * @return
     */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到相应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在，.queueName = " + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            //把消费者 添加到 队列中
            queue.addConsumerEnv(consumerEnv);
            //如果当前队列中已经有了一些消息了，需要立即就消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                //这个方法 ，调用一次就消费一条消息
                consumerMessage(queue);
            }
        }
    }

    /**
     * 消费消息
     *
     * @param queue
     * @return
     */
    private void consumerMessage(MSGQueue queue) {
        //1. 先按照 轮询 的方式，找个消费者出来
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 当前队列没有消费者，暂时不消费，等后面消费者出现再说
            return;
        }

        //2.从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            //当前队列中还没有消息，也不需要消费
            return;
        }
        //3. 把消息带入到消费者的回调方法中，留给线程池执行
        workerPool.submit(() -> {
            try {
                //万一回调方法 没有真正的处理完！ 消息不丢失
                //1. 将这个消息放进待确认的集合中，这个是在回调之前，
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //2. 执行回调
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                //3.autoAck = true 直接把消息删掉
                //  autoAck = false ，则先不处理，之后 消费者调用 basicAck 执行删掉
                if (luckyDog.isAutoAck()) {
                    //1）删除硬盘上的消息 , 2为持久化
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2） 删除待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    //3） 删除内存中心的消息中心的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                }
            } catch (Exception e) {
                e.printStackTrace();

            }

            System.out.println("[ConsumerManager] 消息被成功消费！ queueName = " + queue.getName());
        });


    }
}
